package com.dxy.lp.controller;

import com.alibaba.fastjson.JSON;
import com.dxy.lp.utils.MD5Utils;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationListener;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.AsyncContext;
import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@RestController
public class ServerController implements ApplicationListener<ServerController.ConfigUpdateEvent> {

    public static final String PAIR_SEPARATOR = "@@";
    public static final String WORD_SEPARATOR = ",";
    public static final int LONG_POLLING_TIMEOUT = 30000;// 长轮询等待超时时间默认30s
    public static final int INTERVAL = 500;// 长轮询提前500ms返回避免超时报错

    // 普通线程池
    private ExecutorService executorService = Executors.newFixedThreadPool(3);

    // 长轮询调度线程池
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);

    // 模拟配置表 dataId --> configInfo
    private Map<String, ConfigInfo> configMap = new ConcurrentHashMap<>();

    // 高并发线程安全非阻塞队列
    private Queue<LongPollingTask> taskQueue = new ConcurrentLinkedQueue<>();

    @Autowired
    private ApplicationEventPublisher publisher;

    @GetMapping("test0")
    public void test0(HttpServletRequest request, HttpServletResponse response) throws IOException {
        System.out.println("do business start ...");
        response.setContentType("text/plain;charset=UTF-8");
        response.setStatus(HttpServletResponse.SC_OK);
        response.getWriter().println("hello http");
    }

    @GetMapping("test1")
    public void test1(HttpServletRequest request, HttpServletResponse response) throws IOException, InterruptedException {
        final String name = request.getParameter("name");
        System.out.println("do business start ...");
        TimeUnit.SECONDS.sleep(10);
        System.out.println("do business end ...");
        response.setContentType("text/plain;charset=UTF-8");
        response.setStatus(HttpServletResponse.SC_OK);
        response.getWriter().println("hello, " + name);
        // 在响应完成前，该线程资源都不会被释放。也就是说，处理HTTP请求和执行具体业务代码的线程是同一个线程！
    }

    // ★★★此种做法错误★★★
    // response不能放在异步线程中返回，必须是在主线程中操作，详见test3
    @GetMapping("test2")
    public void test2(HttpServletRequest request, HttpServletResponse response) throws IOException {
        final String name = request.getParameter("name");
        executorService.execute(new BusinessTask0(name, response));
    }

    // 使用Callable接口等待异步任务返回
    @GetMapping("test3")
    public void test3(HttpServletRequest request, HttpServletResponse response) throws IOException, ExecutionException, InterruptedException {
        final String name = request.getParameter("name");
        System.out.println("do business start ...");
        final Future<String> future = scheduledExecutorService.submit(new BusinessTask1(name));
        final String result = future.get();// 此处主线程会阻塞等待，不会释放
        System.out.println("do business end ...");
        response.setContentType("text/plain;charset=UTF-8");
        response.setStatus(HttpServletResponse.SC_OK);
        response.getWriter().println(result);
    }


    // ★★★Servlet3.0后，提供的异步处理新特性★★★
    // ★★★和test3中Callable场景区别是，http线程会释放返回容器！！可以被分配去处理其他请求！！★★★
    // 使Servlet线程不再一直阻塞，直到业务处理完成才能输出响应，最后结束Servlet线程。
    // 在接受到请求之后，Servlet线程可以将耗时的操作委派给一个【异步线程】执行
    // Servlet线程在【不生成响应】的情况下返回容器。这种方案可以大大减少服务器资源占用，提高并发处理速度。
    @GetMapping("test4")
    public void test4(HttpServletRequest request, HttpServletResponse response) throws IOException {
        final String name = request.getParameter("name");
        // ★★★下面这行代码一定要http线程（也就是主线程）去调用，否则会立即返回容器
        final AsyncContext asyncContext = request.startAsync(request, response);
        asyncContext.setTimeout(0);// 这里的超时不准，所以取消掉
        asyncContext.addListener(new MyAsyncListener());// 添加监听器（可选）
        scheduledExecutorService.execute(new BusinessTask2(asyncContext, name));// 异步执行并响应
        System.out.println("此处http线程不会阻塞，可以做其他事情...");// 在用户看来还是会阻塞（等待异步响应）
    }

    // 模拟Nacos更新配置
    // curl -X POST -d "dataId=cipos-work&content=xxx" http://localhost:8080/saveConfig
    @PostMapping("saveConfig")
    public String saveConfig(String dataId, String content) {
        configMap.put(dataId, new ConfigInfo(dataId, content));
        publisher.publishEvent(new ConfigUpdateEvent(this, dataId));
        return "ok";
    }

    // 通过长轮询获取变更的配置信息
    @GetMapping("pullConfigByLongPolling")
    public void pullConfigByLongPolling(HttpServletRequest request, HttpServletResponse response) throws IOException {
        // 解析请求参数，主要是保存在客户端的dataId和md5
        final List<Pair> dataIdAndMd5Pairs = resolveRequestParam(request);

        // 检查当前的配置表是否已经有发生改变的，有就立刻返回
        final List<ConfigInfo> updatedConfigs = checkHasUpdatedConfig(dataIdAndMd5Pairs);

        if (updatedConfigs.size() > 0) {
            response.setContentType("text/plain;charset=UTF-8");
            response.setStatus(HttpServletResponse.SC_OK);
            response.getWriter().println(JSON.toJSONString(updatedConfigs));

        } else {
            // 开启异步机制
            final AsyncContext asyncContext = request.startAsync(request, response);
            asyncContext.setTimeout(0);

            // 放入队列
            final LongPollingTask task = new LongPollingTask(asyncContext, dataIdAndMd5Pairs);
            taskQueue.add(task);

            // 定时29.5s后调度
            long timeout = LONG_POLLING_TIMEOUT - INTERVAL;
            scheduledExecutorService.schedule(task, timeout, TimeUnit.MILLISECONDS);
        }

    }

    // 检查是否有已经变更了的配置
    private List<ConfigInfo> checkHasUpdatedConfig(List<Pair> dataIdAndMd5Pairs) {
        List<ConfigInfo> list = new ArrayList<>();
        for (final Pair pair : dataIdAndMd5Pairs) {
            ConfigInfo configInfo = configMap.get(pair.getDataId());
            if (configInfo == null) {
                // 已删除或者未创建此配置的时候，往里面放一个
                configMap.put(pair.getDataId(), new ConfigInfo(pair.getDataId(), pair.getMd5(), null));
            } else if (!pair.getMd5().equals(configInfo.getMd5())) {
                // md5不相等表示已变更过
                list.add(configInfo);
            }
        }
        return list;
    }

    // 获取客户端请求参数
    private List<Pair> resolveRequestParam(HttpServletRequest request) {
        // dataId,md5|dataId,md5
        String data = request.getParameter("data");
        if (data == null || data.length() == 0) {
            throw new IllegalArgumentException("请求参数不合法");
        }
        System.out.println("pull config request param: " + data);
        final String[] pairs = data.split(PAIR_SEPARATOR);
        List<Pair> list = new ArrayList<>();
        for (String pair : pairs) {
            if (!pair.contains(WORD_SEPARATOR)) {
                throw new IllegalArgumentException("请求参数不合法");
            }
            final String dataId = pair.split(WORD_SEPARATOR)[0];
            final String md5 = pair.split(WORD_SEPARATOR)[1];
            list.add(new Pair(dataId, md5));
        }
        return list;
    }

    @Override
    public void onApplicationEvent(final ConfigUpdateEvent event) {
        final String dataId = event.getDataId();
        System.out.println("配置发生变更，dataId: " + dataId);
        // 采用异步处理，提高并发量
        executorService.execute(new ConfigChangeTask(dataId));
    }

    @AllArgsConstructor
    class BusinessTask0 implements Runnable {

        private String name;
        private HttpServletResponse response;

        @Override
        public void run() {
            try {
                System.out.println("do business start ...");
                TimeUnit.SECONDS.sleep(10);
                System.out.println("do business end ...");
                response.setContentType("text/plain;charset=UTF-8");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println("hello, " + name);
            } catch (InterruptedException | IOException e) {
                e.printStackTrace();
            }
        }
    }

    @AllArgsConstructor
    class BusinessTask1 implements Callable<String> {

        private String name;

        @Override
        public String call() throws Exception {
            System.out.println("... doing something ...");
            TimeUnit.SECONDS.sleep(10);
            return "hello, " + name;
        }
    }

    @AllArgsConstructor
    class BusinessTask2 implements Runnable {

        private AsyncContext asyncContext;
        private String name;

        @Override
        public void run() {
            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
            try {
                System.out.println("do something start ...");
                TimeUnit.SECONDS.sleep(10);
                System.out.println("do something end ...");
                response.setContentType("text/plain;charset=UTF-8");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println("hello, " + name);
                asyncContext.complete();
            } catch (Exception ex) {
                asyncContext.complete();
            }

        }
    }

    @Getter
    @AllArgsConstructor
    class LongPollingTask implements Runnable {

        private AsyncContext asyncContext;
        private List<Pair> pairList;


        @Override
        public void run() {

            // 29.5s已到执行，没有配置变更返回空
            taskQueue.remove(LongPollingTask.this);
            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
            try {
                response.setContentType("text/plain;charset=UTF-8");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println("");
                asyncContext.complete();
            } catch (Exception ex) {
                asyncContext.complete();
            }

        }
    }

    @AllArgsConstructor
    class ConfigChangeTask implements Runnable {

        private String dataId;

        @Override
        public void run() {

            // 遍历长轮询任务队列
            for (Iterator<LongPollingTask> iter = taskQueue.iterator(); iter.hasNext(); ) {
                LongPollingTask task = iter.next();
                if (hasUpdatedConfig(task.getPairList(), dataId)) {
                    iter.remove();

                    final ConfigInfo configInfo = configMap.get(dataId);
                    List<ConfigInfo> list = Arrays.asList(configInfo);

                    final AsyncContext asyncContext = task.getAsyncContext();
                    HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
                    try {
                        response.setContentType("text/plain;charset=UTF-8");
                        response.setStatus(HttpServletResponse.SC_OK);
                        response.getWriter().println(JSON.toJSONString(list));
                        asyncContext.complete();
                    } catch (Exception ex) {
                        asyncContext.complete();
                    }
                }
            }
        }

        // 判断客户端请求的dataId列表有无配置的更新过的
        private boolean hasUpdatedConfig(List<Pair> pairList, String dataId) {
            for (Pair pair : pairList) {
                if (dataId.equals(pair.getDataId())) {
                    // 再判断一下啥都没改重复提交的情况
                    final ConfigInfo lastConfig = configMap.get(dataId);
                    if (!lastConfig.getMd5().equals(pair.getMd5())) {
                        return true;
                    }
                }
            }
            return false;
        }
    }

    class MyAsyncListener implements AsyncListener {

        @Override
        public void onComplete(final AsyncEvent asyncEvent) throws IOException {
            System.out.println("onComplete");
        }

        @Override
        public void onTimeout(final AsyncEvent asyncEvent) throws IOException {
            System.out.println("onTimeout");
        }

        @Override
        public void onError(final AsyncEvent asyncEvent) throws IOException {
            System.out.println("onError");
        }

        @Override
        public void onStartAsync(final AsyncEvent asyncEvent) throws IOException {
            System.out.println("onStartAsync");
        }
    }

    @Getter
    class ConfigUpdateEvent extends ApplicationEvent {

        private String dataId;// 配置文件名


        ConfigUpdateEvent(final Object source, final String dataId) {
            super(source);
            this.dataId = dataId;
        }

    }

    // 配置信息
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    static class ConfigInfo {
        private String dataId;
        private String md5;
        private String content;

        ConfigInfo(final String dataId, final String content) {
            this.dataId = dataId;
            this.content = content;
            this.md5 = MD5Utils.md5(content);
        }

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private static class Pair {
        private String dataId;
        private String md5;
    }

}
